package com.itc.bi.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.itc.bi.bo.DataSetDirctTreeBO;
import com.itc.bi.constant.DataDirectorConstant;
import com.itc.bi.constant.RedisConstant;
import com.itc.bi.constant.SystemConstant;
import com.itc.bi.dto.DirectoryDto;
import com.itc.bi.dto.ReNameDirectDto;
import com.itc.bi.entity.*;
import com.itc.bi.mapper.DataProcessMapper;
import com.itc.bi.mapper.TableMapper;
import com.itc.bi.service.*;
import com.itc.bi.mapper.DirectoryMapper;
import com.itc.bi.vo.Direcr.DircetDataInfoVO;
import com.itc.bi.vo.Direcr.DirectTreeVO;
import com.itc.bi.vo.Direcr.PreviewDatVO;
import com.itc.bi.vo.excelVo.HeardVO;
import com.itc.common.core.exception.ServiceException;
import com.itc.common.core.utils.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author hecai
 * @description 针对表【tb_directory(数据目录表)】的数据库操作Service实现
 * @createDate 2024-06-14 15:56:56
 */
@Service
public class DirectoryServiceImpl extends ServiceImpl<DirectoryMapper, DirectoryEntity> implements DirectoryService {

    @Autowired
    @Lazy
    private DirectDataSetService directDataSetService;
    @Autowired
    private DataSetParameterService dataSetParameterService;
    @Autowired
    private ClickHouseJDBCService clickHouseJDBCService;
    @Autowired
    private DataSetColumnService dataSetColumnService;
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    @Lazy
    private DataSynrecordService dataSynrecordService;
    @Autowired
    private XxlJobInfoService xxlJobInfoService;
    @Autowired
    DataProcessMapper dataProcessMapper;
    @Autowired
    private TableMapper tableMapper;
    @Autowired
    @Lazy
    private DataSetXxlBobService dataSetXxlBobService;

    /**
     * 手动执行一次数据同步
     * @param dtId
     */
    @Override
    public Boolean runOneDataSet(Long dtId){
        //类型判断 DB PYTHON SQL执行同步一次 其他不行
        DirectDataSetEntity entity = directDataSetService.getById(dtId);
        if(ObjectUtil.isEmpty(entity)){
            throw new ServiceException("数据集不存在");
        }
        //excel退出
        switch (entity.getType()){
            case SystemConstant.DATA_SET_TYPE_DB:
                //数据库执行同步一次
                List<TableEntity> tableEntities = tableMapper.selectList(Wrappers.<TableEntity>lambdaQuery()
                        .eq(TableEntity::getDatasourceId, entity.getDsId())
                        .eq(TableEntity::getTableName, entity.getTableName()));
                if(tableEntities!=null && tableEntities.size()>0){
                    TableEntity tableEntity = tableEntities.get(0);
                    dataSynrecordService.execOneTableSyn(entity.getDtId(),tableEntity.getId());
                }
                break;
            case SystemConstant.DATA_SET_TYPE_PYTHON:
                //python执行同步一次
                dataSetXxlBobService.processPythonData(dtId);
                break;
            case SystemConstant.DATA_SET_TYPE_SQL:
                //sql执行同步一次
                dataSetXxlBobService.processSqlData(dtId);
                break;
            default:
                break;
        }
        return true;
    }

    /**
     * 根据数据集id获取cron
     * @param dtId
     * @return
     */
    public String getCronByDtId(Long dtId){
        DirectoryEntity entity = this.getById(dtId);
        if(ObjectUtil.isEmpty(entity)){
            throw new ServiceException("数据集不存在");
        }
        DataSynrecordEntity dataSynrecordEntity = dataSynrecordService.getOne(Wrappers.<DataSynrecordEntity>lambdaQuery().eq(DataSynrecordEntity::getDtId, dtId));
        return dataSynrecordEntity.getCron();
    }

    /**
     * 根据数据集id更新定时任务
     * @param dtId
     * @param cron
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public Boolean updateCron(Long dtId,String cron){
        if(StringUtils.isEmpty(cron)){
            throw new ServiceException("cron不能为空");
        }
        DirectoryEntity entity = this.getById(dtId);
        if(ObjectUtil.isEmpty(entity)){
            throw new ServiceException("数据集不存在");
        }
        DataSynrecordEntity dataSynrecordEntity = dataSynrecordService.getOne(Wrappers.<DataSynrecordEntity>lambdaQuery().eq(DataSynrecordEntity::getDtId, dtId));
        if(ObjectUtil.isNotEmpty(dataSynrecordEntity)){
            dataSynrecordEntity.setCron(cron);
            if(dataSynrecordService.updateById(dataSynrecordEntity)){
                //更新定时任务
                XxlJobInfoEntity xxlJobInfoEntity = xxlJobInfoService.getById(dataSynrecordEntity.getTaskId());
                xxlJobInfoService.stop(dataSynrecordEntity.getTaskId());
                xxlJobInfoEntity.setScheduleConf(cron);
                xxlJobInfoService.updateById(xxlJobInfoEntity);
                CompletableFuture.runAsync(()->{
                    //等待主线程执行完成 数据库事务解开
                    try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    xxlJobInfoService.start(xxlJobInfoEntity.getId());
                }, Executors.newSingleThreadScheduledExecutor());
                return true;
            }
        }
        return false;
    }

    /**
     * 新增目录
     *
     * @param dto
     * @return
     */
    @Override
    public JSONObject saveDirectory(DirectoryDto dto) {
        if (StringUtils.isEmpty(dto.getName())) {
            throw new ServiceException("目录名称不能为空");
        }
        if(ObjectUtil.isNotEmpty(dto.getPid()) && dto.getPid()!=0L){
            long count = this.count(Wrappers.<DirectoryEntity>lambdaQuery()
                    .eq(DirectoryEntity::getDtId, dto.getPid())
                    .eq(DirectoryEntity::getIsDir, DataDirectorConstant.ISDIRECTOR));
            if (count == 0){
                throw new ServiceException("目录不存在");
            }

            long countName = this.count(Wrappers.<DirectoryEntity>lambdaQuery()
                    .eq(DirectoryEntity::getPid, dto.getPid())
                    .likeRight(DirectoryEntity::getName, dto.getName())
                    .eq(DirectoryEntity::getIsDir, DataDirectorConstant.ISDIRECTOR));
            if (countName>0) {
                dto.setName(dto.getName()+countName);
            }
        }
        DirectoryEntity directoryEntity = BeanUtil.copyProperties(dto, DirectoryEntity.class);
        directoryEntity.setIsDir(DataDirectorConstant.ISDIRECTOR);
        //判断是否是存在上层
        if (ObjectUtils.isEmpty(dto.getPid())) {
            directoryEntity.setPid(0L);
        }
        save(directoryEntity);
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("dtId", directoryEntity.getDtId());
        jsonObject.put("pid",directoryEntity.getPid());
        return jsonObject;
    }

    /**
     * 重命名目录
     *
     * @param dto
     * @return
     */
    @Override
    public Boolean reNameDirectory(ReNameDirectDto dto) {
        DirectoryEntity entity = this.getById(dto.getDtId());
        if (ObjectUtil.isEmpty(entity)) {
            throw new ServiceException("目录或数据集不存在");
        }
        entity.setName(dto.getName());
        //TODO 此处还差修改excel表注释逻辑
        return this.updateById(entity);
    }

    /**
     * 移除目录或数据集
     *
     * @param dtId
     * @return
     */
    @Override
    public Boolean reMoveDirectory(Long dtId) {
        DirectoryEntity entity = this.getById(dtId);
        if (ObjectUtil.isEmpty(entity)) {
            throw new ServiceException("目录或数据集不存在");
        }
        LambdaQueryWrapper<DirectoryEntity> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(DirectoryEntity::getPid, dtId);
        if (this.count(queryWrapper) > 0) {
            throw new ServiceException("该目录下存在子目录或数据集，请先删除子目录或数据集");
        }
        if (Objects.equals(entity.getIsDir(), DataDirectorConstant.ISDATASET)) {

            List<DataProcessEntity> dataProcessList = dataProcessMapper.selectList(Wrappers.lambdaQuery(DataProcessEntity.class).eq(DataProcessEntity::getDtId, dtId));
            if(!CollectionUtils.isEmpty(dataProcessList)){
                throw new ServiceException("数据集下有加工数据,请先删除加工数据集");
            }

            //删除数据集
            directDataSetService.remove(Wrappers.<DirectDataSetEntity>lambdaQuery().eq(DirectDataSetEntity::getId, dtId));
            //删除定时任务
            dataSynrecordService.removeByDtId(dtId);
        }
        return this.removeById(dtId);
    }

    /**
     * 复制数据集
     * 复制数据集不考虑数据同步情况
     * @param dtId
     * @return
     */
    @Override
    public Long copyDataSet(Long dtId) {
        DirectoryEntity entity = this.getById(dtId);
        if (ObjectUtil.isEmpty(entity)) {
            throw new ServiceException("目录或数据集不存在");
        }
        //判定只能复制数据集
        if (entity.getIsDir().equals(DataDirectorConstant.ISDIRECTOR)) {
            throw new ServiceException("该数据集是目录，不能复制");
        }
        entity.setDtId(IdWorker.getId());
        entity.setName(entity.getName() + "copy");

        //复制数据集详情
        DirectDataSetEntity directDataSetEntity = this.directDataSetService.getById(dtId);
        if(ObjectUtil.isNotEmpty(directDataSetEntity)){
            directDataSetEntity.setId(entity.getDtId());
            this.directDataSetService.saveOrUpdate(directDataSetEntity);
        }

        //赋值数据项
        List<DataSetColumnEntity> list = this.dataSetColumnService.list(Wrappers.<DataSetColumnEntity>lambdaQuery().eq(DataSetColumnEntity::getDataSetId, dtId).orderByAsc(DataSetColumnEntity::getColumnPosition));
        if(CollectionUtil.isNotEmpty(list)){
            for (DataSetColumnEntity dataSetColumnEntity : list){
                dataSetColumnEntity.setId(IdWorker.getId());
                dataSetColumnEntity.setDataSetId(entity.getDtId());
            }
            this.dataSetColumnService.saveBatch(list);
        }

        //赋值数仓里面数据库表 同一个数据集可能被赋值多次
        String tableName = "copy_"+entity.getDtId();
        String sql = "select * from "+entity.getTableName();
        CompletableFuture.runAsync(()->{
            clickHouseJDBCService.createTableProcessDS(tableName,dtId,sql);
        }, Executors.newSingleThreadScheduledExecutor());
        entity.setTableName(tableName);
        this.save(entity);
        return entity.getDtId();
    }

    /**
     * 数据集移动目录
     * @param dtId  数据集id
     * @param pid   目录id
     * @return
     */
    @Override
    public Boolean mvDirectory(Long dtId, Long pid) {
        DirectoryEntity entity = this.getById(dtId);
        if (ObjectUtil.isEmpty(entity)) {
            throw new ServiceException("数据集不存在");
        }
        if (Objects.equals(entity.getIsDir(), DataDirectorConstant.ISDIRECTOR)) {
            throw new ServiceException("该数据集是目录，不能移动");
        }
        DirectoryEntity directoryEntity = this.getById(pid);
        if (ObjectUtil.isEmpty(directoryEntity)) {
            throw new ServiceException("目录不存在");
        }
        if (Objects.equals(directoryEntity.getIsDir(), DataDirectorConstant.ISDATASET)) {
            throw new ServiceException("该目录是数据集，不能移动");
        }
        entity.setPid(pid);
        return this.updateById(entity);
    }

    /**
     * 获取目录+数据集树结构
     * @return
     */
    @Override
    public  List<DirectTreeVO> getDirectDataSetTree(String name){
//        LambdaQueryWrapper<DirectoryEntity> queryWrapper = new LambdaQueryWrapper<DirectoryEntity>();
//        queryWrapper.like(StringUtils.isNotEmpty(name),DirectoryEntity::getName,name);
//        queryWrapper.orderByAsc(DirectoryEntity::getIsDir);
//        queryWrapper.orderByAsc(DirectoryEntity::getCreateTime);
//        List<DirectoryEntity> list = this.list(queryWrapper);
        List<DataSetDirctTreeBO> dataSetTree = this.getBaseMapper().getDataSetTree();
        return getDirectTreeList(0L,dataSetTree);
    }

    /**
     * 数据预览
     * @param dtId
     * @param pageIndex SELECT dt_id,p_id,name,table_name,version,create_time,update_time,create_by,update_by,is_dir,is_deleted FROM tb_directory WHERE dt_id=? AND is_deleted=0
     * @param pageSize
     * @return
     */
    public PreviewDatVO dataPreviewByDtId(Long dtId,Integer pageIndex,Integer pageSize){
        DirectoryEntity entity = this.getById(dtId);
        if(ObjectUtil.isEmpty(entity) || entity.getIsDir().equals(DataDirectorConstant.ISDIRECTOR) || StringUtils.isEmpty(entity.getTableName())){
            throw new ServiceException("数据集不存在");
        }
        List<LinkedHashMap<String, Object>> hashMaps = this.clickHouseJDBCService.queryTableByTable(entity.getTableName(), pageIndex, pageSize, "");
        List<HeardVO> heardList = dataSetColumnService.heardListById(dtId);
        PreviewDatVO previewDatVO = new PreviewDatVO();
        previewDatVO.setHeardVos(heardList);
        previewDatVO.setBodyVos(hashMaps);
        previewDatVO.setTotalSize(clickHouseJDBCService.getTableSizeTableName(entity.getTableName()));
        return previewDatVO;
    }

    /**
     * 获取预览数据表头
     * @param dtId
     * @return
     */
    public List<HeardVO> getHeardList(Long dtId){
        String redisStr = redisTemplate.opsForValue().get(RedisConstant.DATA_SET_PREVIEW_HEARD + dtId);
        if(StringUtils.isNotEmpty(redisStr)){
            return JSON.parseArray(redisStr,HeardVO.class);
        }
        List<HeardVO> heardVOList = new ArrayList<>();
        List<DataSetColumnEntity> list = dataSetColumnService.list(Wrappers.<DataSetColumnEntity>lambdaQuery().eq(DataSetColumnEntity::getDataSetId, dtId)
                .orderByAsc(DataSetColumnEntity::getColumnPosition));
        for (DataSetColumnEntity directDataSetEntity : list){
            heardVOList.add(new HeardVO(directDataSetEntity.getColumnName(),directDataSetEntity.getColumnComment(),directDataSetEntity.getColumnType()));
        }
        if(ObjectUtil.isNotEmpty(heardVOList)){
            redisTemplate.opsForValue().set(RedisConstant.DATA_SET_PREVIEW_HEARD+dtId, JSON.toJSONString(heardVOList), 120, TimeUnit.SECONDS);
        }
        return heardVOList;
    }

    /**
     * 获取目录树结构
     * @return
     */
    @Override
    public List<DirectTreeVO> getDirectTree(){
        return getDirectTree(0L,this.list());
    }

    public List<DirectTreeVO> getDirectTree(Long pid,List<DirectoryEntity> list){
        List<DirectTreeVO> children = new ArrayList<>();
        for (DirectoryEntity entity : list){
            DirectTreeVO vo = new DirectTreeVO();
            if (Objects.equals(entity.getPid(), pid)){
                vo.setChildren(getDirectTree(entity.getDtId(),list));
                vo.setDtId(entity.getDtId());
                vo.setName(entity.getName());
                vo.setIsDir(entity.getIsDir().equals(DataDirectorConstant.ISDIRECTOR)?true:false);
                vo.setPid(entity.getPid());
                //数据集不能加入
                if(vo.getIsDir()){
                    children.add(vo);
                }
            }
        }
        return children;
    }

    /**
     * 获取目录树结构
     * @param pid
     * @param list
     * @return
     */
    public List<DirectTreeVO> getDirectTreeList(Long pid,List<DataSetDirctTreeBO> list){
        List<DirectTreeVO> children = new ArrayList<>();
        for (DataSetDirctTreeBO entity : list){
            DirectTreeVO vo = new DirectTreeVO();
            if (Objects.equals(entity.getPid(), pid)){
                vo.setChildren(getDirectTreeList(entity.getDtId(),list));
                vo.setDtId(entity.getDtId());
                vo.setType(entity.getType());
                vo.setIsEditCron(StringUtils.isNotEmpty(entity.getCron())?true:false);
                vo.setName(entity.getName());
                vo.setIsDir(entity.getIsDir().equals(DataDirectorConstant.ISDIRECTOR)?true:false);
                vo.setPid(entity.getPid());
                children.add(vo);
            }
        }
        return children;
    }


    /**
     * 获取数据集详情
     * @param dtId
     * @return
     */
    public DircetDataInfoVO getDircetDataInfo(Long dtId){
        DircetDataInfoVO dircetDataInfo = super.baseMapper.getDircetDataInfo(dtId);
        if(ObjectUtil.isNotEmpty(dircetDataInfo)){
            String tableName = dircetDataInfo.getTableName();
            dircetDataInfo.setDataTotal(clickHouseJDBCService.getTableSizeTableName(tableName));
        }
        return dircetDataInfo;
    }

    /**
     * 验证字段类型是否可转换三部
     * 1.强制  时间和数值  数值和时间
     * 2.类型级别
     * 3.数据级别
     * @param id
     * @param targetType
     * @return
     */
    @Override
    public Boolean isTransitionColumType(Long id,String targetType){
        DataSetColumnEntity entity = this.dataSetColumnService.getById(id);
        String sourceType = entity.getColumnType();
        if(sourceType.equalsIgnoreCase(targetType)){
            return true;
        }
        //1.强制  时间和数值  数值和时间不能转换
        if(sourceType.equalsIgnoreCase("number")&&targetType.equalsIgnoreCase("timestamp")){
            return false;
        }
        if(sourceType.equalsIgnoreCase("timestamp")&&targetType.equalsIgnoreCase("number")){
            return false;
        }
        Long dtId = entity.getDataSetId();
        DirectoryEntity directoryEntity = this.getById(dtId);
        String tableName = directoryEntity.getTableName();
        String sql2 = "select "+buildToTypeColumn(targetType,entity.getColumnName())+" from "+tableName + " limit 1";
        //2.类型级别成功后 在验证 3.数据行级别
        if(clickHouseJDBCService.execSql(sql2)){
            String sql3 = "select `"+entity.getColumnName()+"` from "+tableName + " where  `"+entity.getColumnName()+"` is not null limit 1 ";
            try{
                List<LinkedHashMap<String, Object>> linkedHashMaps = clickHouseJDBCService.queryTableBySql(sql3);
                if(linkedHashMaps!=null && linkedHashMaps.size()>0){
                    LinkedHashMap<String, Object> stringObjectLinkedHashMap = linkedHashMaps.get(0);
                    Object objectData = stringObjectLinkedHashMap.getOrDefault(entity.getColumnName(), "");
                    if(ObjectUtil.isNotEmpty(objectData)){
                        String sql33 = "select "+buildToTypeColumnData(targetType,objectData.toString());
                        if(clickHouseJDBCService.execSql(sql33)){
                            return true;
                        }
                    }
                }
                //没有数据返回 暂时处理成能转换类型
                else {
                    return true;
                }
            }catch (Exception ex){
                return false;
            }
        }
        return false;
    }

    /**
     * 构建转换类型
     * @param columnType
     * @param columnName
     * @return
     */
    private String buildToTypeColumn(String columnType,String columnName){
        switch (columnType){
            case "number":
                return "toFloat64(`"+columnName+"`)";
            case "timestamp":
                return "toDateTime(`"+columnName+"`)";
            case "string":
                return "toString(`"+columnName+"`)";
            default:
                return "`"+columnName+"`";
        }
    }


    /**
     * 构建转换类型
     * @param columnType
     * @param data
     * @return
     */
    private String buildToTypeColumnData(String columnType,String data){
        switch (columnType){
            case "number":
                return "toFloat64('"+data+"')";
            case "date":
                return "toDateTime('"+data+"')";
            case "string":
                return "toString('"+data+"')";
            default:
                return data;
        }
    }
}




